#include <stdio.h>
#include <stdlib.h>
#include <inttypes.h>
#include <string.h>
#include <errno.h>
#include "ursax.h"
#include "log.h"
#include "anysocket.h"
#include "protocol.h"
#include "networking.h"
#include "master-request.h"


uint32_t manager_ip;
uint16_t manager_port = 9000;

int request_manager_create_chunk_location_(st_netfd_t stfd, struct MgrReqChunkLocate* req, struct CSEP* csids)
{
	LOG_DEBUG("%s(volumeid=%08x, chunk_size=%lu, n=%u)", __func__, req->volumeid, size32_convert_to64(req->chunk_size), req->cs_cnt);

	struct MgrRespChunkLocate resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_CONNECTION_ERROR_LOG_RETURN(ret, req, resp);

	uint32_t size = req->cs_cnt * sizeof(*csids);
	ret = st_read_fully(stfd, csids, size, URSAX_TIMEOUT);
	if (unlikely(ret < (int)size))
	{
		LOG_ERROR("%s() st_read_fully failed %d: %s", __func__, errno, strerror(errno));
		return ret;
	}
	return 0;
}

int request_manager_chunkserver_up_(st_netfd_t stfd, struct MgrReqServerUp* req)
{
	LOG_DEBUG("%s(my_ip=%s, my_port=%u)", __func__, str_ip(req->csid.ip), req->csid.port);

	struct MgrRespServerUp resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_report_chunk_error_(st_netfd_t stfd, struct MgrReqChunkError* req)
{
	LOG_DEBUG("%s(cs_ip=%s, cs_port=%u, op=%u, connection_code=%d, operation_code=%d)",
		__func__, str_ip(req->csid.ip), req->csid.port, req->operation, req->connection_error_code, req->operation_error_code);

	struct MgrRespChunkError resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_report_chunk_error2_(st_netfd_t stfd, struct MgrReqChunkError2* req)
{
	LOG_INFO("%s(cs_ip=%s, cs_port=%u, volumeid=%08x, index=%u, op=%u, connection_code=%d, operation_code=%d)",
		__func__, str_ip(req->csid.ip), req->csid.port, req->id.volumeid, req->id.index,
		req->operation, req->connection_error_code, req->operation_error_code);

	struct MgrRespChunkError2 resp;
	int ret = operation_send_recv_with_timeout(stfd, req, sizeof(*req), &resp, sizeof(&resp), 1*(1000*1000));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_report_chunk_error3_(st_netfd_t stfd, struct MgrReqChunkError3* req)
{
	LOG_INFO("%s(cs_ip=%s, cs_port=%u, volumeid=%08x, index=%u, op=%u, connection_code=%d, operation_code=%d)",
		__func__, str_ip(req->csid.ip), req->csid.port, req->id.volumeid, req->id.index,
		req->operation, req->connection_error_code, req->operation_error_code);

	struct MgrRespChunkError3 resp;
	int ret = operation_send_recv_with_timeout(stfd, req, sizeof(*req), &resp, sizeof(&resp), 1*(1000*1000));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_create_volume_(st_netfd_t stfd, struct MgrReqVolumeCreate* req, char* name, char* desc)
{
	LOG_DEBUG("%s name %s desc %s", __func__, name, desc);
	struct MgrRespVolumeCreate resp;
	int len1 = (sizeof(req->header) + sizeof(req->size) + sizeof(req->name));
	int len = st_write(stfd, req, len1, URSAX_TIMEOUT);
	len = st_write(stfd, name, req->name.length, URSAX_TIMEOUT);
	len = st_write(stfd, &req->desc, sizeof(req->desc), URSAX_TIMEOUT);
	len = st_write(stfd, desc, req->desc.length, URSAX_TIMEOUT);
	len = st_write(stfd, &req->data, sizeof(req->data), URSAX_TIMEOUT);
	if(len < (int)sizeof(req->data)){
		LOG_ERROR("error write req");
		return -1;
	}
	flush_tcp(stfd);
	len = st_read_fully(stfd, &resp, sizeof(resp), URSAX_TIMEOUT);
	if(len < (int)sizeof(resp)){
		LOG_ERROR("error read resp");
		return -1;
	}
	CHECK_ERROR_LOG_RETURN(0, req, resp, stfd);
	return resp.volumeid;
}

int request_manager_move_chunk_(st_netfd_t stfd, struct MgrReqChunkMove* req)
{
	struct MgrRespChunkMove resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(resp));
	if(ret < 0){
		LOG_ERROR("error operation send recv %d", ret);
		return ret;
	}
	return 0;
}

int request_manager_open_volume_(st_netfd_t stfd, struct MgrReqVolumeOpen* req, char* clientid, char** jvolume)
{
	LOG_DEBUG("%s(volume-id=%08x, readonly=%u, client-id='%s')", __func__, req->volumeid, req->readonly, clientid);

	struct MgrRespVolumeOpen resp;
	uint32_t n = req->clientid.length = strlen(clientid);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), clientid, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);

	*jvolume = (char*)zalloc(resp.jvolume.length + 1);
	ret = st_read_fully(stfd, *jvolume, resp.jvolume.length, URSAX_TIMEOUT);
	if (ret < 0)
	{
		LOG_ERROR("%s() st_read_fully failed %d: %s", __func__, errno, strerror(errno));
		return ret;
	}
	return 0;
}

int request_manager_open_volume_name_(st_netfd_t stfd, struct MgrReqVolumeOpenName* req, char* name, char* clientid, char** jvolume)
{
	LOG_DEBUG("%s(volume-key=%s, readonly=%u, client-id='%s')", __func__, name, req->readonly, clientid);

	struct MgrRespVolumeOpen resp;
	uint32_t name_len = req->name = strlen(name);
	uint32_t client_len = req->clientid.length = strlen(clientid);

	int total_send_len = 0;
	int len1 = sizeof(req->header) + sizeof(req->byid) + sizeof(req->name);
	int len = st_write(stfd, req, len1, URSAX_TIMEOUT);
	total_send_len += len;
	len = st_write(stfd, name, name_len, URSAX_TIMEOUT);
	total_send_len += len;
	len = st_write(stfd, &req->readonly, sizeof(req->readonly), URSAX_TIMEOUT);
	total_send_len += len;
	len = st_write(stfd, &req->clientid, sizeof(req->clientid), URSAX_TIMEOUT);
	total_send_len += len;
	len = st_write(stfd, clientid, client_len, URSAX_TIMEOUT);
	total_send_len += len;
	if(len != (int)client_len){
		LOG_ERROR("error read resp");
		return -1;
	}
	LOG_DEBUG("name %s clientid %s total sendlen %d", name, clientid, total_send_len);
	flush_tcp(stfd);
	len = st_read_fully(stfd, &resp, sizeof(resp), URSAX_TIMEOUT);
	if(len < (int)sizeof(resp)){
		LOG_ERROR("error read resp");
		return -1;
	}
	CHECK_ERROR_LOG_RETURN(0, req, resp, stfd);
	*jvolume = (char*)zalloc(resp.jvolume.length + 1);
	int ret = st_read_fully(stfd, *jvolume, resp.jvolume.length, URSAX_TIMEOUT);
	if (ret < 0)
	{
		LOG_ERROR("%s() st_read_fully failed %d: %s", __func__, errno, strerror(errno));
		return ret;
	}
	return 0;
}

// todo same as request_manager_open_volume_
int request_manager_open_volumeX_(st_netfd_t stfd, struct MgrReqVolumeOpen2* req, char* clientid, char** jvolume)
{
	LOG_DEBUG("%s(volume-id=%08x, readonly=%u, client-id='%s')", __func__, req->volumeid, req->readonly, clientid);

	MgrRespVolumeOpen2 resp;
	uint32_t n = req->clientid.length = strlen(clientid);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), clientid, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);

	char *j = (char *)zalloc(resp.jvolume.length + 1);
	ret = st_read_fully(stfd, j, resp.jvolume.length, URSAX_TIMEOUT);
	if (ret < 0)
	{
		LOG_ERROR("%s() st_read_fully failed %d: %s", __func__, errno, strerror(errno));
		free(j);
		return ret;
	}

	if(jvolume){
		*jvolume = j;
	}else{
		free(j);
	}

	return 0;
}

int request_manager_close_volume_(st_netfd_t stfd, struct MgrReqVolumeClose* req, char* clientid)
{
	LOG_DEBUG("%s(volume-id=%08x, client-id='%s')", __func__, req->volumeid, clientid);

	struct MgrRespVolumeClose resp;
	uint32_t n = req->clientid.length = strlen(clientid);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), clientid, n, &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_resize_volume_(st_netfd_t stfd, struct MgrReqVolumeResize* req)
{
	LOG_DEBUG("%s(volume-id=%08x, new_size=%llu, preallocation=%u)", __func__, req->volumeid, req->new_size, req->preallocation);

	struct MgrRespVolumeResize resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_volume_setformat_(st_netfd_t stfd, struct MgrReqVolumeSetfmt *req)
{
	LOG_DEBUG("%s(volume-id=%08x, new_fmt=%u, virtual-size=%llu)", __func__, req->volumeid, req->new_fmt, req->virtual_size);

	struct MgrRespVolumeSetfmt resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_client_fault_(st_netfd_t stfd, struct MgrReqClientFault *req, char* clientid)
{
	LOG_DEBUG("%s ", __func__);
	struct MgrRespClientFault resp;
	uint32_t n = req->clientid.length = strlen(clientid);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), clientid, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_snapshot_AddAndUpdate_(st_netfd_t stfd, struct MgrReqSnapshotAdd *req, char *json_str)
{
	LOG_INFO("%s %s", __func__, json_str);
	struct MgrRespSnapshotAdd resp;
	uint32_t n= req->msg.length = strlen(json_str);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), json_str, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_snapshot_delete_(st_netfd_t stfd, struct MgrReqSnapshotDelete *req)
{
	LOG_INFO("%s %x %d", __func__, req->volume_id, req->id);
	struct MgrRespSnapshotDelete resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(&resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_snapshot_get_(st_netfd_t stfd, struct MgrReqSnapshotGet *req, char **msg_out)
{
	LOG_DEBUG("%s ", __func__);
	struct MgrRespSnapshotGet resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	*msg_out = (char *)zalloc(resp.msg.length + 1);
	ret = st_read_fully(stfd, *msg_out, resp.msg.length, URSAX_TIMEOUT);
	if(ret < (int)resp.msg.length){
		LOG_ERROR("error get snapshot on mysql");
		return -1;
	}
	return 0;
}

int request_manager_volume_set_active_(st_netfd_t stfd, struct MgrReqVolumeSetActive *req)
{
	LOG_DEBUG("%s ", __func__);
	struct MgrRespVolumeSetActive resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_backup_AddAndUpdate_(st_netfd_t stfd, struct MgrReqBackupAdd *req, char *json_str)
{
	LOG_DEBUG("%s (%s)", __func__, json_str);
	struct MgrRespBackupAdd resp;
	uint32_t n = req->msg.length = strlen(json_str);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), json_str, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_backup_delete_(st_netfd_t stfd, struct MgrReqBackupDelete *req, char *name)
{
	LOG_DEBUG("%s ", __func__);
	struct MgrRespBackupDelete resp;
	uint32_t n = req->name.length = strlen(name);
	int ret = operation_send2_recv(stfd, req, sizeof(*req), name, n, &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	return 0;
}

int request_manager_backup_get_(st_netfd_t stfd, struct MgrReqBackupGet *req, char **msg_out)
{
	LOG_DEBUG("%s ", __func__);
	struct MgrRespBackupGet resp;
	int ret = operation_send_recv(stfd, req, sizeof(*req), &resp, sizeof(resp));
	CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd);
	*msg_out = (char *)zalloc(resp.msg.length + 1);
	ret = st_read_fully(stfd, *msg_out, resp.msg.length, URSAX_TIMEOUT);
	if(ret < (int)resp.msg.length){
		LOG_ERROR("error get backup on mysql");
		return -1;
	}
	LOG_DEBUG("get backups (%s)", *msg_out);
	return 0;
}
